package cn.trigram.uuid.bd.buffer;

import cn.trigram.asserts.AssertUtil;
import cn.trigram.uuid.bd.utils.NamingThreadFactory;
import cn.trigram.uuid.bd.utils.PaddedAtomicLong;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 代表一个填充{@link RingBuffer} 的 executor
 * <br/>
 * 存在两种 executor ：一种是计划填充，另一种是立即填充
 */
public class BufferPaddingExecutor {

  private static final Logger LOGGER = LoggerFactory.getLogger(RingBuffer.class);

  /** Constants */
  private static final String WORKER_NAME = "RingBuffer-Padding-Worker";

  private static final String SCHEDULE_NAME = "RingBuffer-Padding-Schedule";

  // 默认计划执行间隔为 5分钟
  private static final long DEFAULT_SCHEDULE_INTERVAL = 5 * 60L;

  /**
   * Whether buffer padding is running
   * <br/>
   * buffer 填充是否运行
   * */
  private final AtomicBoolean running;

  /**
   * We can borrow UIDs from the future, here store the last second we have consumed
   * <br/>
   * 从未来借用UIDs，这里我们存储了消费的最后一秒的时间值
   * */
  private final PaddedAtomicLong lastSecond;

  /** RingBuffer & BufferUidProvider */
  private final RingBuffer ringBuffer;

  private final BufferedUidProvider uidProvider;

  /**
   * Padding immediately by the thread pool
   * <br/>
   * 通过线程池立即填充，固定线程池，核心数的两倍
   * */
  private final ExecutorService bufferPadExecutors;

  /**
   * Padding schedule thread
   * <br/>
   * 计划执行线程池，单线程
   * */
  private final ScheduledExecutorService bufferPadSchedule;

  /**
   * Schedule interval Unit as seconds
   * <br/>
   * 计划执行线程的间隔时间
   * */
  private long scheduleInterval = DEFAULT_SCHEDULE_INTERVAL;

  /**
   * Constructor with {@link RingBuffer} and {@link BufferedUidProvider}, default use schedule
   *
   * @param ringBuffer
   *     {@link RingBuffer}
   * @param uidProvider
   *     {@link BufferedUidProvider}
   */
  public BufferPaddingExecutor(RingBuffer ringBuffer, BufferedUidProvider uidProvider) {

    this(ringBuffer, uidProvider, true);
  }

  /**
   * Constructor with {@link RingBuffer}, {@link BufferedUidProvider}, and whether use schedule
   * padding
   *
   * @param ringBuffer
   *     {@link #ringBuffer}
   * @param uidProvider
   *     {@link #uidProvider}
   * @param usingSchedule
   *     是否使用计划填充
   */
  public BufferPaddingExecutor(RingBuffer ringBuffer, BufferedUidProvider uidProvider,
      boolean usingSchedule) {

    this.running = new AtomicBoolean(false);
    this.lastSecond = new PaddedAtomicLong(
        TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()));
    this.ringBuffer = ringBuffer;
    this.uidProvider = uidProvider;

    // initialize thread pool
    int cores = Runtime.getRuntime().availableProcessors();
    bufferPadExecutors = Executors
        .newFixedThreadPool(cores * 2, new NamingThreadFactory(WORKER_NAME));

    // initialize schedule thread
    if (usingSchedule) {
      bufferPadSchedule = Executors
          .newSingleThreadScheduledExecutor(new NamingThreadFactory(SCHEDULE_NAME));
    } else {
      bufferPadSchedule = null;
    }
  }

  /**
   * 开始计划执行填充缓存
   */
  public void start() {

    if (bufferPadSchedule != null) {
      bufferPadSchedule
          .scheduleWithFixedDelay(this::paddingBuffer, scheduleInterval, scheduleInterval,
              TimeUnit.SECONDS);
    }
  }

  /**
   * Shutdown executors
   */
  public void shutdown() {

    if (!bufferPadExecutors.isShutdown()) {
      bufferPadExecutors.shutdownNow();
    }

    if (bufferPadSchedule != null && !bufferPadSchedule.isShutdown()) {
      bufferPadSchedule.shutdownNow();
    }
  }

  /**
   * Whether is padding
   */
  public boolean isRunning() {

    return running.get();
  }

  /**
   * Padding buffer in the thread pool
   * 立刻异步执行填充缓存
   */
  public void asyncPadding() {

    bufferPadExecutors.submit(this::paddingBuffer);
  }

  /**
   * Padding buffer fill the slots until to catch the cursor
   * 填充缓存直到填充满
   */
  public void paddingBuffer() {

    LOGGER.info("Ready to padding buffer lastSecond:{}. {}", lastSecond.get(), ringBuffer);

    // is still running
    if (!running.compareAndSet(false, true)) {
      LOGGER.info("Padding buffer is still running. {}", ringBuffer);
      return;
    }

    // fill the rest slots until to catch the cursor
    boolean isFullRingBuffer = false;
    while (!isFullRingBuffer) {
      List<Long> uidList = uidProvider.provide(lastSecond.incrementAndGet());
      for (Long uid : uidList) {
        isFullRingBuffer = !ringBuffer.put(uid);
        if (isFullRingBuffer) {
          break;
        }
      }
    }

    // not running now
    running.compareAndSet(true, false);
    LOGGER.info("End to padding buffer lastSecond:{}. {}", lastSecond.get(), ringBuffer);
  }

  /**
   * Setters
   */
  public void setScheduleInterval(long scheduleInterval) {

    AssertUtil.assertTrue(scheduleInterval > 0, "Schedule interval must positive!");
    this.scheduleInterval = scheduleInterval;
  }

}
